package com.chencong.online.driver.dwd;

import com.chencong.online.bean.UniqueVisitorBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.UVCountAllWindowFunc;
import com.chencong.online.function.UserBehaviorBeanMapFunc;
import com.chencong.online.utils.StarRocksUtil;
import com.starrocks.connector.flink.row.StarRocksSinkRowBuilder;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import java.net.URL;
import java.time.Duration;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdUniqueVisitorDriver
 * @description:UV计算
 * @author: chencong
 * @create: 2021-12-28 15:17
 **/
public class DwdUniqueVisitorDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //todo 1 source
        URL resource = DwdPageViewDriver.class.getResource("/UserBehavior.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print("获取原始数据");
        //todo 2 transform
        //转换层javabean并提取时间戳
        SingleOutputStreamOperator<UserBehaviorBean> userBehaviorDS = inputDS
                .map(new UserBehaviorBeanMapFunc())
                .filter(data -> "pv".equals(data.getBehavior()))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehaviorBean>() {
                            @Override
                            public long extractTimestamp(UserBehaviorBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        }));
        //开窗统计UV
        SingleOutputStreamOperator<UniqueVisitorBean> resultDS = userBehaviorDS
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1))) //(全窗口函数不按照key分组)将所有数据发送到下游单个实例。下游算子的并行度为1
                .apply(new UVCountAllWindowFunc());//源码：此函数不提供增量聚合方法
        resultDS.print("去重后的数据");
        //todo 3 sink：starRocks

        // the table structure（目标表的结果和字段）
        TableSchema tableSchema = TableSchema.builder()
                .field("window_time", DataTypes.VARCHAR(256))
                .field("uv", DataTypes.BIGINT())
                .build();
        // set the slots with streamRowData（要输出的数据）
        StarRocksSinkRowBuilder<UniqueVisitorBean> starRocksSinkRowBuilder = new StarRocksSinkRowBuilder<UniqueVisitorBean>() {
            @Override
            public void accept(Object[] objects, UniqueVisitorBean uniqueVisitorBean) {
                objects[0] = uniqueVisitorBean.getWindowEnd();
                objects[1] = uniqueVisitorBean.getCount();

            }
        };
        //获取sink函数
        SinkFunction<UniqueVisitorBean> uniqueVisitorBeanSinkFunction = StarRocksUtil.sinkStarRocks
                (tableSchema, "ads_unique_visitor_count", "ads", starRocksSinkRowBuilder);
        //写入starRocks
        resultDS.addSink(uniqueVisitorBeanSinkFunction);

        env.execute("DwdUniqueVisitorDriver");
    }
}
